package com.chb.flink.time

;

import java.text.SimpleDateFormat
import java.util.Date

import com.chb.flink.source.StationLog
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

class ReturnMaxTimeWindowFunction extends WindowFunction[StationLog, String, String, TimeWindow] {
    //在窗口触发的才调用一次
    override def apply(key: String, window: TimeWindow, input: Iterable[StationLog], out: Collector[String]): Unit = {
        var value = input.iterator.next()
        var sb = new StringBuilder
        val df = new SimpleDateFormat("HH:mm:ss")
        sb.append("窗口范围是:").append(df.format(new Date(window.getStart))).append("----").append(df.format(new Date(window.getEnd)))
        sb.append(", 基站id:").append(value.sid)
        sb.append(", 呼叫时间：").append(value.callTime)
            .append(", 主叫号码：").append(value.callOut)
            .append(", 被叫号码：").append(value.callIn)
            .append(", 通话时长：").append(value.duration)
            .append("\n")
        out.collect(sb.toString())
    }
}

